package cn.doitedu.jobs;

import cn.doitedu.pojo.LogBean;
import cn.doitedu.pojo.MatchResult;
import cn.doitedu.udfs.JsonToLogBeanFunction;
import cn.doitedu.udfs.RulesMatchingKeyedProcessFunction;
import cn.doitedu.utils.FlinkUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @create: 2021-11-12 18:44
 * @author: 静如海的男人
 * @program: eagle-text
 **/

public class RulesMatching {
    public static void main(String[] args) throws Exception {
        DataStream<String> kafkaStream = FlinkUtils.createKafkaStream(args[0], SimpleStringSchema.class);

        SingleOutputStreamOperator<LogBean> beanStream = kafkaStream.process(new JsonToLogBeanFunction());

        SingleOutputStreamOperator<LogBean> watermarks = beanStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LogBean>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(LogBean element) {
                return element.getTimeStamp();
            }
        });

        KeyedStream<LogBean, String> keyedStream = watermarks.keyBy(LogBean::getDeviceId);

        SingleOutputStreamOperator<MatchResult> res = keyedStream.process(new RulesMatchingKeyedProcessFunction());

        res.print();

        FlinkUtils.env.execute();
    }
}
